package Demo3;

import com.google.common.base.Optional;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import scala.Tuple2;

import java.util.Arrays;
import java.util.List;

/**
 * Created by lenovo on 2017/12/7.
 *   * updateStateByKey算子：在流式的处理中记住之前的值进行累计，该方法会保存上次计算数据的状态，以供下次计算使用
 */
public class Streaming_updateStateByKeyJava {
    public static void main(String[] args){
        System.setProperty("hadoop.home.dir", "E://hadoop-liyadong//hadoop-2.7.1");

        SparkConf conf = new SparkConf().setMaster("local[2]").setAppName("Streaming_updateStateByKeyJava")
                .set("spark.testing.memory","2147480000");
//        JavaSparkContext sc = new JavaSparkContext(conf);
//        new StreamingContext(sc, Seconds(5));
        JavaStreamingContext ssc = new JavaStreamingContext(conf, Durations.seconds(5));
        ssc.checkpoint(".");
       JavaReceiverInputDStream linesDStream = ssc.socketTextStream("hadoop1",2211);

       JavaDStream<String> flatMapDStream = linesDStream.flatMap(new FlatMapFunction<String,String>() {
            @Override
            public Iterable call(String line) throws Exception {
                return Arrays.asList(line.split(" "));
            }
        });

      JavaPairDStream<String,Integer> wordsDStream = flatMapDStream.mapToPair(new PairFunction<String, String, Integer>() {

            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s,1);
            }
        });

      JavaPairDStream<String,Integer> wordDSteam =  wordsDStream.updateStateByKey(new Function2<List<Integer>, Optional<Integer>, Optional<Integer>>() {
            @Override
            public Optional<Integer> call(List<Integer> v1, Optional<Integer> v2) throws Exception {
                Integer newValue = 0;
                if(v2.isPresent()){
                    newValue = v2.get();
                }
                for(Integer value:v1){
                    newValue += value;
                }

                return Optional.of(newValue);
            }
        });

        wordDSteam.print();

        ssc.start();
        ssc.awaitTermination();
        ssc.close();


    }
}
